package io.gitee.zhangbinhub.acp.boot.websocket.reactive

import io.gitee.zhangbinhub.acp.boot.exceptions.WebSocketException
import io.gitee.zhangbinhub.acp.boot.log.LogAdapter
import io.gitee.zhangbinhub.acp.boot.tools.SslTools
import org.springframework.http.HttpHeaders
import org.springframework.web.reactive.socket.WebSocketSession
import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.net.URI
import java.util.concurrent.ConcurrentHashMap
import javax.net.ssl.SSLContext

abstract class AcpWebSocketClient
@JvmOverloads constructor(
    logAdapter: LogAdapter,
    @JvmField protected val clientId: String,
    @JvmField protected val sslType: String = "SSL"
) : AcpWebSocketHandlerReactive(logAdapter) {
    @JvmOverloads
    @Throws(WebSocketException::class)
    fun connect(
        uri: URI,
        headers: HttpHeaders? = null
    ): Mono<Void> = close().then(Mono.defer {
        ReactorNettyWebSocketClient().let { webSocketClient ->
            if (uri.scheme == "wss") {
                SSLContext.getInstance(sslType)?.apply {
                    this.init(null, arrayOf(SslTools.getTrustManager()), null)
                    SSLContext.setDefault(this)
                }
            }
            if (headers == null) {
                webSocketClient.execute(uri, this)
            } else {
                webSocketClient.execute(uri, headers, this)
            }
        }
    })

    fun close(): Mono<Void> = getInstance(clientId)?.session?.close() ?: Mono.empty()

    override fun handle(session: WebSocketSession): Mono<Void> {
        this.session = session
        acpWebSocketInstanceMap[clientId] = this
        val input = doReceiveMessage(session)
        val output = session.send(Flux.create { fluxSink ->
            this.outFluxSink = fluxSink
            sessionOpen(session)
        })
        return Flux.merge(input, output).doFinally {
            acpWebSocketInstanceMap.remove(clientId)
            sessionClose(session)
        }.then()
    }

    companion object {
        private val acpWebSocketInstanceMap: ConcurrentHashMap<String, AcpWebSocketClient> =
            ConcurrentHashMap<String, AcpWebSocketClient>()

        @JvmStatic
        fun getInstance(clientId: String): AcpWebSocketClient? = acpWebSocketInstanceMap[clientId]

        @JvmStatic
        fun getInstanceList(): List<AcpWebSocketClient> = acpWebSocketInstanceMap.map { entry -> entry.value }
    }

}